package com.hps.shoping.good.component;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.hps.shoping.good.service.GoodService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.List;

/**
 * @author heps
 * @date 2019/9/1 10:30
 */
@Component
@Slf4j
public class MessageConsumer {

    DefaultMQPushConsumer consumer;

    private final GoodService goodService;

    private final MessageProducer messageProducer;

    public MessageConsumer(GoodService goodService, MessageProducer messageProducer) {
        this.goodService = goodService;
        this.messageProducer = messageProducer;
        consumer = new DefaultMQPushConsumer("order_group");
        consumer.setNamesrvAddr("192.168.117.139:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        try {
            consumer.subscribe("orderTopic", "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt msg : msgs) {
                        String body = new String(msg.getBody(), Charset.defaultCharset());
                        JSONObject order = JSON.parseObject(body);
                        log.info("update stock {}", order);

                        Long goodId = order.getLong("goodId");
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("orderId", order.getString("id"));
                        try {
                            boolean result = goodService.updateStock(order.getString("id"), goodId);
                            if (result) {
                                jsonObject.put("status", 1);
                            } else {
                                jsonObject.put("status", 2);
                            }
                        } catch (Exception e) {
                            jsonObject.put("status", 2);
                        }
                        try {
                            messageProducer.send(jsonObject.toJSONString());
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }
}
